#include "event_loop.h"
#include "logger.h"
#include <sys/eventfd.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>

namespace net {

// EventLoop 实现
EventLoop::EventLoop() 
    : _quit(false)
    , _threadId(std::this_thread::get_id())
    , _poller(new Poller())
    , _eventFd(createEventFd())
    , _eventChannel(new Channel(this, _eventFd))
    , _timerWheel(this) 
{
    // 注册eventfd的读事件回调（用于唤醒事件循环）
    _eventChannel->setReadCallback(std::bind(&EventLoop::readEventfd, this));
    _eventChannel->enableReading();
    LOG_INFO("Thread %d: EventLoop created", std::this_thread::get_id());
}

EventLoop::~EventLoop() 
{
    _eventChannel->disableAll(); // 给Channel移除所有感兴趣的事件
    _eventChannel->remove();     // 把Channel从EventLoop上删除掉
    LOG_INFO("Thread %d: eventFd: %d", std::this_thread::get_id(), _eventFd); 
    if(_eventFd >= 0)
    {
        LOG_INFO("Thread %d: eventFd destroyed", std::this_thread::get_id()); 
        close(_eventFd);
    }
        
    LOG_INFO("Thread %d: EventLoop destroyed", std::this_thread::get_id());
}

void EventLoop::start() 
{
    LOG_INFO("Thread %d: EventLoop starting", std::this_thread::get_id());
    while (!_quit) 
    {
        // 1. 监控事件（获取就绪的Channel）
        std::vector<Channel*> actives;
        _poller->poll(&actives);

        // 2. 处理就绪事件
        for (auto& channel : actives) 
        {
            // 这里的事件一般是读写事件就绪，也就是需要的IO操作
            channel->handleEvent();
        }

        // 3. 执行任务队列
        runAllTask();
    }
    LOG_INFO("Thread %d: EventLoop stop.", std::this_thread::get_id());
}

void EventLoop::runInLoop(const Functor& cb) 
{
    if (isInLoop()) 
    {
        // 是当前线程的任务，直接处理，避免性能损失
        cb();
    } 
    else 
    {
        // 不是当前线程，直接将这个任务压入到目标loop中的任务队列中
        // 后续loop->start中会处理这些任务，确保了在对应线程中执行，避免线程安全问题
        queueInLoop(cb);
    }
}

void EventLoop::queueInLoop(const Functor& cb) 
{
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _tasks.emplace_back(cb);
    }
    /**
     * 当前loop可能正在执行回调中 但是loop的_task中又加入了新的回调 
     * 需要通过wakeup写事件唤醒相应的需要执行上面回调操作的loop的线程 
     * 然后继续执行_taks中的回调函数
     **/
    // 唤醒事件循环（避免epoll阻塞）
    wakeUpEventFd();
}

void EventLoop::updateEvent(Channel* channel) 
{
    _poller->updateEvent(channel);
}

void EventLoop::removeEvent(Channel* channel) 
{
    _poller->removeEvent(channel);
}

/**
 * 退出事件循环
 * 1. 如果loop在自己的线程中调用quit成功了 说明当前线程已经执行完毕了start()函数的_poller.poll()并退出
 * 2. 如果不是当前EventLoop所属线程中调用quit退出EventLoop 需要唤醒EventLoop所属线程的epoll_wait
 *
 * 比如在一个subloop(worker)中调用mainloop(IO)的quit时 需要唤醒mainloop(IO)的poller_->poll 让其执行完loop()函数
 *
 * ！！！ 注意： 正常情况下 mainloop负责请求连接 将回调写入subloop中 通过生产者消费者模型即可实现线程安全的队列
 * ！！！       但是muduo通过wakeup()机制 使用eventfd notify 使得mainloop和subloop之间能够进行通信
 **/
void EventLoop::quit() 
{
    _quit = true;
    if (!isInLoop()) 
    {
        wakeUpEventFd();  // 若在其他线程调用quit，需唤醒当前线程
    }
}

void EventLoop::runAllTask() 
{
    std::vector<Functor> tasks;
    // 批量取出任务（减少锁持有时间）
    {
        std::unique_lock<std::mutex> lock(_mutex);
        tasks.swap(_tasks);
    }

    // 执行任务
    for (const auto& task : tasks) 
    {
        task();
    }
}

int EventLoop::createEventFd() 
{
    int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    if (efd < 0) 
    {
        LOG_FATAL("Thread %d: Create eventfd failed! error: %s", 
                 std::this_thread::get_id(), strerror(errno));
    }
    LOG_INFO("EventFd created successfully. FD: %d", efd);
    return efd;
}

void EventLoop::readEventfd() 
{
    uint64_t val = 0;
    ssize_t ret = read(_eventFd, &val, sizeof(val));
    if (ret < 0) 
    {
        if (errno != EINTR && errno != EAGAIN) 
        {
            LOG_ERROR("Thread %d: Read eventfd failed! error: %s", 
                     std::this_thread::get_id(), strerror(errno));
            LOG_FLUSH();
        }
    }
}

void EventLoop::wakeUpEventFd() 
{
    uint64_t val = 1;
    ssize_t ret = write(_eventFd, &val, sizeof(val));
    if (ret < 0) 
    {
        if (errno != EINTR) 
        {
            LOG_FATAL("Thread %d: Write eventfd failed! error: %s", 
                     std::this_thread::get_id(), strerror(errno));
            LOG_FLUSH();
        }
    }
}

// LoopThread 实现
LoopThread::LoopThread() 
    : _eventLoop(nullptr), 
      _thread(std::thread(&LoopThread::threadEntry, this))
{
    LOG_INFO("Thread %d: LoopThread created", std::this_thread::get_id());
}

LoopThread::~LoopThread() 
{
    if (_eventLoop) 
    {
        _eventLoop->quit();
        if (_thread.joinable()) 
        {
            _thread.join();
        }
    }
    
    LOG_INFO("Thread %d: LoopThread destroyed", std::this_thread::get_id());
}

EventLoop* LoopThread::getLoop() 
{
    EventLoop *loop = nullptr;
    {
        std::unique_lock<std::mutex> lock(_mutex);
        // 等待EventLoop初始化完成
        _cond.wait(lock, [this]() { return _eventLoop != nullptr; });
        loop = _eventLoop;
    }
    return loop;
}

void LoopThread::threadEntry() 
{
    EventLoop loop; // 在新线程中创建EventLoop
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _eventLoop = &loop;
        _cond.notify_all(); // 通知getLoop()等待完成
    }
    loop.start(); // 启动事件循环（阻塞）

    // 确保在退出前清理资源
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _eventLoop = nullptr;
    }
}

// LoopThreadPool 实现
LoopThreadPool::LoopThreadPool(EventLoop* baseLoop) 
    : _threadCount(0)
    , _nextIdx(0)
    , _baseLoop(baseLoop) 
{
    LOG_INFO("Thread %d: LoopThreadPool created", std::this_thread::get_id());
}

LoopThreadPool::~LoopThreadPool() 
{
    LOG_INFO("Thread %d: LoopThreadPool destroyed", std::this_thread::get_id());
}

void LoopThreadPool::create() 
{
    if (_threadCount <= 0) 
    {
        LOG_WARN("Thread %d: LoopThreadPool thread count is 0", std::this_thread::get_id());
        return;
    }

    _threads.reserve(_threadCount);
    _loops.reserve(_threadCount);

    for (int i = 0; i < _threadCount; ++i) 
    {
        _threads.emplace_back(std::unique_ptr<LoopThread>(new LoopThread()));
        _loops.emplace_back(_threads.back()->getLoop());
        LOG_INFO("Thread %d: LoopThreadPool created thread %d", 
                 std::this_thread::get_id(), i);
    }
}

EventLoop* LoopThreadPool::nextLoop() 
{
    if (_threadCount == 0) 
    {
        return _baseLoop;
    }

    // 轮询选择下一个EventLoop
    _nextIdx = (_nextIdx + 1) % _threadCount;
    return _loops[_nextIdx];
}

} // namespace net